﻿using NetMQ;
using NetMQ.Sockets;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace ParallelLib.Micro
{
    public class MyBus : IDisposable
    {
        private readonly NetMQActor m_actor;
        private PublisherSocket m_publisher;
        private SubscriberSocket m_subscriber;
        private NetMQBeacon m_beacon;
        private NetMQPoller m_poller;
        private PairSocket m_shim;
        private NetMQTimer m_mqtimer;

        private readonly TimeSpan m_deadNodeTimeout = TimeSpan.FromSeconds(10);
        private readonly Dictionary<NodeKey, DateTime> m_nodes;
        private readonly int m_broadcastPort;
        private int m_randomPort;
        private string mLastEndpoint;
        private bool disposedValue;

        public Dictionary<NodeKey, DateTime> Nodes
        {
            get
            {
                return m_nodes;
            }
        }

        public string HostAddress { get; private set; }
        public string ID { get; private set; }
        public int PID { get; private set; }

        public MyBus(int port, string id)
        {
            ID = id;
            PID = Environment.ProcessId;
            m_nodes = new Dictionary<NodeKey, DateTime>();
            m_broadcastPort = port;
            m_actor = NetMQActor.Create(RunActor);
            Thread.Sleep(1000);
            m_actor.SendFrame(BusCmd.GetHostAddrCMD);
            var hostAddress = m_actor.ReceiveFrameString();
            HostAddress = hostAddress;
        }

        /// <summary>
        /// Creates a new message bus actor. All communication with the bus is
        /// through the returned <see cref="NetMQActor"/>.
        /// </summary>
        public static NetMQActor Create(int broadcastPort, string id, out MyBus bus)
        {
            bus = new MyBus(broadcastPort, id);
            return bus.m_actor;
        }

        private void RunActor(PairSocket shim)
        {
            // save the shim to the class to use later
            m_shim = shim;
            // create all subscriber, publisher and beacon
            using (m_subscriber = new SubscriberSocket())
            using (m_publisher = new PublisherSocket())
            using (m_beacon = new NetMQBeacon())
            {
                //订阅所有的消息
                m_subscriber.Subscribe("");
                //订阅者绑定一个随机的端口，随后使用 beacon发布该端口
                m_randomPort = m_subscriber.BindRandomPort("tcp://*");
                mLastEndpoint = m_subscriber.Options.LastEndpoint;
                //给beacon配置特定的端口
                m_beacon.Configure(m_broadcastPort);
                //使用beacon将随机端口发布到所有的其他节点
                var msgs = string.Format("{0}{1}{2}", m_randomPort, BusCmd.SplitChar, ID);
                //m_beacon.Publish(m_randomPort.ToString(), TimeSpan.FromSeconds(1));
                m_beacon.Publish(msgs, TimeSpan.FromSeconds(1));
                //beacon订阅所有的其他已配置特定端口的beacon的所有的UDP消息
                m_beacon.Subscribe("");
                //创建一个用于清理下线的结点的定时器
                NetMQTimer timer = new NetMQTimer(TimeSpan.FromSeconds(1));
                m_mqtimer = timer;
                //创建一个轮询器
                m_poller = new NetMQPoller();
                m_poller.Add(m_shim);
                m_poller.Add(m_subscriber);
                m_poller.Add(m_beacon);
                m_poller.Add(timer);
                //监听所有的actor的命令
                m_shim.ReceiveReady += OnShimReady;
                //监听其他发布者发送的消息，并推送给shim
                m_subscriber.ReceiveReady += OnSubscriberReady;
                //监听所有新加入的beacon
                m_beacon.ReceiveReady += OnBeaconReady;
                //定时器时间到执行下线节点的清理
                timer.Elapsed += ClearDeadNodes;
                //告诉actor已完成配置开始执行
                m_shim.SignalOK();
                //开始轮询直到取消
                m_poller.Run();
            }
        }

        #region 服务发现相关事件

        /// <summary>
        /// 清理下线节点
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void ClearDeadNodes(object sender, NetMQTimerEventArgs e)
        {
            var deadNodes = m_nodes
                .Where(n => DateTime.Now > n.Value + m_deadNodeTimeout)
                .Select(n => n.Key).ToArray();
            foreach (var node in deadNodes)
            {
                m_nodes.Remove(node);
                m_publisher.Disconnect(node.Address);
                m_shim.SendMoreFrame(BusCmd.DelNodeCMD).SendFrame(node.Address);
            }
        }

        /// <summary>
        /// 有新节点加入
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnBeaconReady(object sender, NetMQBeaconEventArgs e)
        {
            var message = m_beacon.Receive();
            if (string.IsNullOrEmpty(message.String))
                return;
            var msgs = message.String.Split(BusCmd.SplitChar);
            if (msgs.Length != 2)
                return;
            int port;
            int.TryParse(msgs[0], out port);
            string id = msgs[1];

            NodeKey node = new NodeKey(message.PeerHost, port, id);
            if (!m_nodes.ContainsKey(node))
            {
                //缓存新节，连接新节点
                m_nodes.Add(node, DateTime.Now);
                m_publisher.Connect(node.Address);
                //发布新加入节点消息，公布新节点地址
                m_shim.SendMoreFrame(BusCmd.AddNodeCMD).SendFrame(node.Address);
            }
            else
            {
                //更新已有节点的存活时间
                m_nodes[node] = DateTime.Now;
            }
        }

        /// <summary>
        /// 监听其他发布者发送的消息，并推送给shim
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnSubscriberReady(object sender, NetMQSocketEventArgs e)
        {
            NetMQMessage message = m_subscriber.ReceiveMultipartMessage();
            m_shim.SendMultipartMessage(message);
        }

        /// <summary>
        /// 监听所有的actor的命令
        /// </summary>
        /// <param name="sender"></param>
        /// <param name="e"></param>
        private void OnShimReady(object sender, NetMQSocketEventArgs e)
        {
            string command = m_shim.ReceiveFrameString();
            switch (command)
            {
                case NetMQActor.EndShimMessage:
                    // we cancel the socket which dispose and exist the shim
                    m_poller.Stop();
                    break;

                case BusCmd.PublishCMD:
                    // it is a publish command we just forward everything to the publisher until end of message
                    NetMQMessage message = m_shim.ReceiveMultipartMessage();
                    m_publisher.SendMultipartMessage(message);
                    break;

                case BusCmd.GetHostAddrCMD:
                    var address = m_beacon.BoundTo + ":" + m_randomPort;
                    m_shim.SendFrame(address);
                    break;

                default:
                    NetMQMessage msg = m_shim.ReceiveMultipartMessage();
                    Console.WriteLine("收到：：\t" + msg.ToString());
                    break;
            }
        }

        #endregion 服务发现相关事件

        public void SayTo(NodeKey key, string msg)
        {
            if (key == null)
                return;
            var keystr = string.Format("{0}_{1}", key.ID, key.Name);
            var frame = new NetMQMessage();
            frame.Append(BusCmd.RecvMsgCMD);
            frame.Append(keystr + msg, System.Text.Encoding.UTF8);
            m_publisher.SendMultipartMessage(frame);
            //m_actor.SendMoreFrame(BusCmd.RecvMsgCMD).SendMoreFrame(keystr).SendFrame(msg);
        }

        public void Publish(string msg)
        {
            var frame = new NetMQMessage();
            frame.Append(BusCmd.PublishCMD);
            frame.Append(msg, System.Text.Encoding.UTF8);
            m_publisher.SendMultipartMessage(frame);
            //m_actor.SendMoreFrame(BusCmd.PublishCMD).SendMoreFrame(msg).SendFrame(HostAddress);
        }

        public string RecvMsg()
        {
            return m_actor?.ReceiveFrameString();
        }

        public void Stop()
        {
            m_actor.SendMoreFrame(NetMQActor.EndShimMessage);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    // TODO: 释放托管状态(托管对象)
                    m_poller.Stop();
                    m_actor.Dispose();
                    m_publisher.Dispose();
                    m_beacon.Dispose();
                    m_shim.Dispose();
                    m_subscriber.Dispose();
                    m_mqtimer.Enable = false;
                }
                // TODO: 释放未托管的资源(未托管的对象)并重写终结器
                // TODO: 将大型字段设置为 null
                disposedValue = true;
            }
        }

        // // TODO: 仅当“Dispose(bool disposing)”拥有用于释放未托管资源的代码时才替代终结器
        // ~MyBus()
        // {
        //     // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        //     Dispose(disposing: false);
        // }

        public void Dispose()
        {
            // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
            Dispose(disposing: true);
            GC.SuppressFinalize(this);
        }
    }
}